package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSqlApp;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/24 11:13
 */
public class DWSProvinceStatsApp extends BaseSqlApp {
    public static void main(String[] args) {
        new DWSProvinceStatsApp().init(30003, 2, "DWSProvinceStatsApp");
    }
    
    @Override
    public void run(StreamTableEnvironment tenv) {
        // 1. 建表与source关联 t1 dwm_order_wide
        tenv.executeSql("create table order_wide(" +
                            "   province_id bigint, " +
                            "   province_name string, " +
                            "   province_area_code string, " +
                            "   province_iso_code string, " +
                            "   province_3166_2_code string, " +
                            "   order_id bigint, " +
                            "   split_total_amount decimal(12, 2), " +
                            "   create_time string, " +
                            "   event_time as TO_TIMESTAMP(create_time), " +
                            "   watermark for event_time as event_time - interval '5' second" +
                            ")with(" +
                            "   'connector'='kafka', " +
                            "   'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092,hadoop164:9092'," +
                            "   'properties.group.id' = 'DWSProvinceStatsApp'," +
                            "   'topic' = 'dwm_order_wide'," +
                            "   'scan.startup.mode' = 'latest-offset'," +
                            "   'format' = 'json'" +
                            ")");
        
        // 2. 建表与sink关联  t2
        tenv.executeSql("create table province_stats_2021(" +
                            "   stt string," +
                            "   edt string," +
                            "   province_id bigint," +
                            "   province_name string," +
                            "   area_code string," +
                            "   iso_code string," +
                            "   iso_3166_2 string," +
                            "   order_amount decimal(20, 2)," +
                            "   order_count bigint," +
                            "   ts bigint, " +
                            "   primary key(stt, edt, province_id) not enforced " +
                            ")with(" +
                            "   'connector' = 'clickhouse',  " +
                            "   'url' = 'clickhouse://hadoop162:8123',  " +
                            "   'database-name' = 'gmall2021', " +
                            "   'table-name' = 'province_stats_2021'," +
                            "   'sink.batch-size' = '100', " +
                            "   'sink.flush-interval' = '1000', " +
                            "   'sink.max-retries' = '3' " +
                            ")");
    
        // 3. 查询t1, 把结果写入到t2
        final Table table = tenv.sqlQuery("select " +
                                              " date_format(tumble_start(event_time, interval '10' second), 'yyyy-MM-dd HH:mm:ss'), " +
                                              " date_format(tumble_end(event_time, interval '10' second), 'yyyy-MM-dd HH:mm:ss'), " +
                                              " province_id, " +
                                              " province_name, " +
                                              " province_area_code, " +
                                              " province_iso_code, " +
                                              " province_3166_2_code, " +
                                              " sum(split_total_amount) order_amount, " +
                                              " count(distinct order_id) order_count, " +
                                              " UNIX_TIMESTAMP()*1000 ts " +
                                              "from order_wide " +
                                              "group by " +
                                              " tumble(event_time, interval '10' second)," +
                                              " province_id, " +
                                              " province_name, " +
                                              " province_area_code, " +
                                              " province_iso_code, " +
                                              " province_3166_2_code ");
//        table.execute().print();
        
        table.executeInsert("province_stats_2021");
        
    
    }
}
